package com.atguigu.flinkSql;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.e;

/**
 * @author wky
 * @create 2021-07-21-11:10
 */
public class Flink03_TableApi_File_source {
    public static void main(String[] args) {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(environment);
        // TODO 2.1表的元数据信息
        Schema schema = new Schema()
                .field("id", DataTypes.STRING())
                .field("ts", DataTypes.BIGINT())
                .field("vc", DataTypes.INT());
        //TODO 2.2 连接文件 创建一个临时表 指定参数
        tableEnvironment
                .connect(new FileSystem()
                //连接数据位置
                .path("src/input/sensor_sql.txt"))
                //指定分割
                .withFormat(new Csv().fieldDelimiter(',').lineDelimiter("\n"))
                //指定元数据
                .withSchema(schema)
                //创建临时表 sensor
                .createTemporaryTable("sensor");
        //TODO 2.3 做成表对象 然后查询
        Table table = tableEnvironment.from("sensor")
                .groupBy($("id"))
                .select($("id"), $("id").count().as("count"));
        //sql写法  打印
//        Table table1 = tableEnvironment.sqlQuery("select * from sensor");
//        TableResult result1 = table1.execute();
//        result1.print();
        //通过调用execute这个方法返回一个TableResult类型可以直接打印 不需要 env执行
        TableResult result = table.execute();
        result.print();

//        将table 转化成 数据流输出 这个需要执行 env
//        DataStream<Tuple2<Boolean, Row>> dataStream = tableEnvironment.toRetractStream(table, Row.class);
//        dataStream.print();
//        try {
//            environment.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
    }

}
